//和网络相关的函数

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <stdarg.h>
#include <unistd.h>
#include <sys/time.h>
#include <time.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/ioctl.h>
#include <arpa/inet.h>

#include "configure.h"
#include "macro.h"
#include "global.h"
#include "function.h"
#include "mysocket.h"
#include "mymemory.h"
#include "lock_mutex.h"


//构造函数
MySocket::MySocket() {
	//成员变量默认值初始化
	worker_connections_ = 1;//epoll最大连接数
	listen_port_count_ = 1;
	recy_connection_wait_time_ = 60;

	pkg_head_len_ = sizeof(COMM_PKG_HEADER);
	msg_head_len_ = sizeof(STRUC_MSG_HEADER);

	//各种队列
	send_msg_queue_count_ = 0;
	totol_recy_connections_ = 0;
	current_size_ = 0;//当前计时器队列大小
	timer_value_  = 0;

	discard_send_pkg_count_ = 0;

	//在线用户相关
	online_user_count_ = 0;
	last_print_time_ = 0;

	return;
}

//初始化函数
bool MySocket::Initialize() {
	ReadConf();
	if(OpenListeningSocket() == false) {
		return false;
	}
	return true;
}
//在子进程中的初始化函数
bool MySocket::InitializeSubproc() {
	
	//互斥量初始化
	if(pthread_mutex_init(&send_message_queue_mutex_, NULL) != 0) {
		LogStderr(0, "MySocket::InitializeSubproc()中pthread_mutex_init(&send_message_queue_mutex_)失败.");
		return false;
	}
	if(pthread_mutex_init(&connection_mutex_, NULL) != 0) {
		LogStderr(0, "MySocket::InitializeSubproc()中pthread_mutex_init(&connection_mutex_)失败.");
		return false;
	}
	if(pthread_mutex_init(&recy_connect_queue_mutex_, NULL) != 0) {
		LogStderr(0, "MySocket::InitializeSubproc()中pthread_mutex_init(&recy_connect_queue_mutex_)失败.");
		return false;
	}
	if(pthread_mutex_init(&time_queue_mutex_, NULL) != 0) {
		LogStderr(0, "MySocket::InitializeSubproc()中pthread_mutex_init(&time_queue_mutex_)失败.");
		return false;
	}

	
	//初始化信号量
	if(sem_init(&sem_event_send_queue_, 0, 0) == -1) {
		LogStderr(0, "MySocket::InitializeSubproc()中sem_init(&sem_event_send_queue_,0,0)失败.");
		return false;
	}
	//创建线程
	int err;
	ThreadItem *send_queue;
	thread_vector_.push_back(send_queue = new ThreadItem(this));

	err = pthread_create(&send_queue->_handle, NULL, ServerSendQueueThread, send_queue);
	if(err != 0) {
		LogStderr(0, "MySocket::InitializeSubProc()中pthread_create(ServerSendQueueThread)失败.");
		return false;
	}

	//专门用于回收连接的线程
	ThreadItem *recy_connect;
	thread_vector_.push_back(recy_connect = new ThreadItem(this));
	err = pthread_create(&recy_connect->_handle, NULL, ServerRecyConnectionThread, recy_connect);
	if(err != 0) {
		LogStderr(0, "MySocket::InitializeSubproc()中pthread_create(ServerRecyConnectionThread)失败.");
		return false;
	}

	//是否开启踢人时钟
	if(if_kick_time_count_ == 1) {
		ThreadItem *time_monitor;
		thread_vector_.push_back(time_monitor = new ThreadItem(this));
		err = pthread_create(&time_monitor->_handle, NULL, ServerTimerQueueMonitorThread, time_monitor);
		if(err != 0) {
			LogStderr(0, "MySocket::InitializeSubproc()中pthread_create(ServerTimerQueueMonitorThread)失败.");
			return false;
		}
	}


	return true;
}


MySocket::~MySocket() {
	//释放监听端口相关内存
	std::vector<lplistening_t>::iterator pos;
	for(pos = listen_socket_list_.begin(); pos != listen_socket_list_.end(); ++pos) {
		delete(*pos);
	}
	listen_socket_list_.clear();
	return;
}

//关闭退出函数
void MySocket::ShutdownSubproc() {
	if(sem_post(&sem_event_send_queue_) == -1) {
		LogStderr(0, "MySocket::ShutdownSubproc()中sem_post(&sem_event_send_queue_)失败.");
	}
	std::vector<ThreadItem *>::iterator iter;
	for(iter = thread_vector_.begin(); iter != thread_vector_.end(); iter++) {
		pthread_join((*iter)->_handle, NULL);
	}
	for(iter = thread_vector_.begin(); iter != thread_vector_.end(); iter++) {
		if(*iter)
			delete *iter;
	}
	thread_vector_.clear();
	//清理队列
	ClearMsgSendQueue();
	ClearConnection();
	ClearAllFromTimerQueue();
	//多线程相关
	pthread_mutex_destroy(&connection_mutex_);
	pthread_mutex_destroy(&send_message_queue_mutex_);
	pthread_mutex_destroy(&recy_connect_queue_mutex_);
	pthread_mutex_destroy(&time_queue_mutex_);
	sem_destroy(&sem_event_send_queue_);
}

void MySocket::ClearMsgSendQueue() {
	char *tmp_mem_point;
	MyMemory *memory = MyMemory::GetInstance();
	
	while(!msg_send_queue_.empty()) {
		tmp_mem_point = msg_send_queue_.front();
		msg_send_queue_.pop_front();
		memory->FreeMemory(tmp_mem_point);
	}
}

//读取各项配置项
void MySocket::ReadConf() {
	Configure *configure = Configure::GetInstance();
	listen_port_count_ = configure->GetIntDefault("ListenPortCount",listen_port_count_);
	worker_connections_ = configure->GetIntDefault("WorkerConnections", worker_connections_);
	recy_connection_wait_time_ = configure->GetIntDefault("SockRecyConnectionWaitTime",recy_connection_wait_time_);//延迟连接回收的时间
	
	if_kick_time_count_ = configure->GetIntDefault("SockWaitTimeEnable", 0);
	wait_time_ = configure->GetIntDefault("SockMaxWaitTime", wait_time_);
	wait_time_  = wait_time_ > 5? wait_time_:5;
	if_timeout_kick_ = configure->GetIntDefault("SockTimeOutKick", 0);
	flood_attack_enable_ = configure->GetIntDefault("SockFloodAttackKickEnable", 0);
	flood_time_interval_ = configure->GetIntDefault("SockFloodTimeInterval", 100);
	flood_kick_count_ = configure->GetIntDefault("SockFloodKickCounter", 10);

	return;
}

//监听端口，在执行worker进程前执行
bool MySocket::OpenListeningSocket() {
	int                sock;
	struct sockaddr_in serv_addr;
	int                port;
	char               str_info[100];

	memset(&serv_addr, 0, sizeof(struct sockaddr_in));
	serv_addr.sin_family = AF_INET;
	serv_addr.sin_addr.s_addr   = htonl(INADDR_ANY);

	Configure *configure = Configure::GetInstance();
	for(int i = 0;i < listen_port_count_; i++) {
		sock = socket(AF_INET, SOCK_STREAM, 0);
		if(sock == -1) {
			LogStderr(errno, "MySocket::Initialize()中socket()失败, i=%d.", i);
			return false;
		}

		int re_use_addr = 1;//是否复用本地址
		if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (const void*)&re_use_addr, sizeof(re_use_addr)) == -1) {
			LogStderr(errno, "MySocket::Initialize()中setsockopt(SO_REUSEADDR)失败, i=%d.", i);
			close(sock);
			return false;
		}
		int re_use_port = 1;//为了处理惊群问题
		if(setsockopt(sock, SOL_SOCKET, SO_REUSEPORT, (const void*)&re_use_port,sizeof(int)) == -1) {
			LogStderr(errno, "MySocket::Initialize()中setsockopt(SO_REUSEPORT)失败", i);
		}

		if(SetNonblocking(sock) == false) {
			LogStderr(errno, "MySocket::Initialize()中SetNonblocking()失败,i=%d.", i);
			close(sock);
			return false;
		}

		str_info[0] = 0;
		sprintf(str_info, "ListenPort%d", i);
		port  = configure->GetIntDefault(str_info, 10000);
		serv_addr.sin_port = htons((in_port_t)port);

		//绑定服务器地址结构体
		if(bind(sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) == -1) {
			LogStderr(errno, "MySocket::Initialize()中bind()失败,i=%d",i);
			close(sock);
			return false;
		}

		if(listen(sock, LISTEN_BACKLOG) == -1) {
			LogStderr(errno, "MySocket::Initialize()中listen()失败,i=%d.", i);
			close(sock);
			return false;
		}
		//LogErrorCore(LOG_INFO, 0, "端口监听成功,%d",port);
		lplistening_t listen_socket_item = new listening_t;
		memset(listen_socket_item, 0, sizeof(listening_t));
		listen_socket_item->port = port;
		listen_socket_item->fd   = sock;
		listen_socket_list_.push_back(listen_socket_item);
	}
	if(listen_socket_list_.size() <= 0) {
		return false;
	}
	return true;
}

//设置socket为非阻塞模式
bool MySocket::SetNonblocking(int sockfd) {
	int nb = 1;
	if(ioctl(sockfd, FIONBIO, &nb) == -1) {
		return false;
	}
	return true;
}

//关闭监听队列
void MySocket::CloseListeningSocket() {
	for(int i=0; i<listen_port_count_; i++) {
		close(listen_socket_list_[i]->fd);
		LogErrorCore(LOG_INFO, 0, "关闭监听端口%d!", listen_socket_list_[i]->port);
	}
	return;
}

//待发送消息入发送消息队列
void MySocket::MsgSend(char *send_buf) {
	//LogStderr(0, "将准备的消息装入发送消息队列中");
	MyMemory *memory = MyMemory::GetInstance();
	Lock lock(&send_message_queue_mutex_);

	if(send_msg_queue_count_ > 50000) {
		//消息过多，选择抛弃
		discard_send_pkg_count_++;
		memory->FreeMemory(send_buf);
		return;
	}

	LPSTRUC_MSG_HEADER msg_header = (LPSTRUC_MSG_HEADER)send_buf;
	lpconnection_t connect = msg_header->connect;
	if(connect->send_count > 400) {
		//用户收数据太慢
		LogStderr(0, "MySocket::MsgSend()中发现某用户%d积压了大量待发送数据包,切断与他的连接!", connect->fd);
		discard_send_pkg_count_++;
		memory->FreeMemory(send_buf);
		ActiveCloseSocketProc(connect);
		return;
	}
	//LogStderr(0, "将待发送的消息装入待发送的消息队列");
	++connect->send_count;
	msg_send_queue_.push_back(send_buf);
	++send_msg_queue_count_;
	//LogStderr(0, "通知专门负责消息发送的线程处理");
	if(sem_post(&sem_event_send_queue_) == -1) {
		LogStderr(0, "MySocket::MsgSend()中sem_post(&sem_event_send_queue_)失败.");
	}
	return;
}

//主动关闭TCP连接
void MySocket::ActiveCloseSocketProc(lpconnection_t connect) {
	//时间队列
	if(connect->fd != -1) {
		close(connect->fd);
		connect->fd = -1;
	}
	if(connect->throw_send_count > 0) {
		--connect->throw_send_count;
	}
	InRecyConnectQueue(connect);
	return;
}


//测试洪范攻击
bool MySocket::TestFlood(lpconnection_t connect) {
	struct timeval current_time_struct;
	uint64_t current_time;
	bool reco = false;

	gettimeofday(&current_time_struct, NULL);
	current_time = (current_time_struct.tv_sec * 1000 + current_time_struct.tv_usec / 1000);

	if((current_time - connect->flood_kick_last_time) < flood_time_interval_) {
		connect->flood_attack_count++;
		connect->flood_kick_last_time = current_time;
	}
	else {
		connect->flood_attack_count = 0;
		connect->flood_kick_last_time = current_time;
	}

	if(connect->flood_attack_count >= flood_kick_count_) {
		reco = true;
	}
	return reco;
}


//打印统计信息
void MySocket::PrintInfo() {
	time_t current_time = time(NULL);
	if((current_time - last_print_time_) > 10) {
		int msg_queue_count = thread_pool.GetRecvMsgQueueCount();
		last_print_time_ = current_time;

		int online_user_count = online_user_count_;
		int send_msg_queue_count = send_msg_queue_count_;
		
		LogStderr(0, "---------------------------begin---------------------------");
		LogStderr(0, "当前在线人数/总人数(%d/%d)",online_user_count, worker_connections_);
		LogStderr(0, "连接池中空闲连接/总连接/要释放的连接(%d/%d/%d)",free_connection_list_.size(), connection_list_.size(), recy_connection_list_.size());
		LogStderr(0, "当前时间队列大小(%d)", time_queue_map_.size());
		LogStderr(0, "当前收消息队列/发消息队列大小分别是(%d/%d),丢弃的待发送数据包数量为%d.",msg_queue_count, send_msg_queue_count, discard_send_pkg_count_);

		if(msg_queue_count > 100000) {
			LogStderr(0, "接收队列条目数量过大(%d),要考虑限速或者增加处理线程数量", msg_queue_count);
		}
		LogStderr(0, "---------------------------end-----------------------------");
	}
}

//epoll功能初始化
int MySocket::EpollInit() {
	epoll_handle_ = epoll_create(worker_connections_);
	if(epoll_handle_ == -1) {
		LogStderr(errno, "MySocket::EpollInit()中epoll_create()失败.");
		exit(2);
	}

	//初始化连接池
	InitConnection();
	std::vector<lplistening_t>::iterator pos;
	for(pos = listen_socket_list_.begin(); pos!=listen_socket_list_.end(); ++pos) {
		lpconnection_t connect = GetConnection((*pos)->fd);
		if(connect == NULL) {
			LogStderr(errno, "MySocket::EpollInit()中GetConnection()失败.");
			exit(2);
		}
		connect->listening = (*pos);
		(*pos)->connection = connect;
		connect->read_handler = &MySocket::EventAccept;
		if(EpollOperEvent((*pos)->fd, EPOLL_CTL_ADD, EPOLLIN|EPOLLRDHUP, 0, connect) == -1) {
		exit(2);
		}
	}
	return 1;
}

//堆epoll事件的具体操作
int MySocket::EpollOperEvent(int fd, uint32_t event_type, uint32_t flag, int supply_action, lpconnection_t connect) {
	struct epoll_event ev;
	memset(&ev, 0, sizeof(ev));

	if(event_type == EPOLL_CTL_ADD) {
		ev.events = flag;
		//ev.events |= EPOLLET;
		connect->events = flag;//？？？连接本身记录这个标记的意义
	}

	else if(event_type == EPOLL_CTL_MOD) {
		ev.events = connect->events;
		if(supply_action == 0) {
			//增加某个标记
			ev.events |= flag;
		}
		else if(supply_action == 1) {
			//去掉某个标记
			ev.events &= ~flag;
		}
		else {
			//完全覆盖某个标记
			ev.events = flag;
		}
		connect->events = ev.events;
	}
	else {
		return 1;
	}
	ev.data.ptr = (void *)connect;
	if(epoll_ctl(epoll_handle_, event_type, fd, &ev) == -1) {
		LogStderr(errno, "MySocket::EpollOperEvent()中epoll_ctl(%d,%ud,%ud,%d)失败.", fd, event_type, flag, supply_action);
		return -1;
	}
	return 1;
}

int MySocket::EpollProcessEvent(int timer) {
	int events = epoll_wait(epoll_handle_, events_, MAX_EVENTS, timer);
	if(events == -1) {
		if(errno == EINTR) {
			LogErrorCore(LOG_INFO, errno, "MySocket::EpollProcessEvents()中epoll_wait()失败!");
			return 1;
		}
		else {
			LogErrorCore(LOG_ALERT, errno, "MySocket::EpollProcessEvents()中epoll_wait()失败!");
			return 0;
		}
	}

	if(events == 0) {
		if(timer != -1) {
			return 1;
		}
		LogErrorCore(LOG_ALERT, 0, "MySocket::EpollProcessEvents()中epoll_wait()没有超时去没有返回任何事件!");
		return 0;
	}

	//处理新来到的事件
	lpconnection_t connect;
	uint32_t revents;
	//LogStderr(0, "处理新事件");
	for(int i=0;i<events;++i) {
		connect = (lpconnection_t)(events_[i].data.ptr);
		revents = events_[i].events;//取出事件
		if(revents & EPOLLIN) {
			//如果是读事件
			(this->*(connect->read_handler))(connect);

		}
		if(revents & EPOLLOUT) {
			//如果是写事件
			if(revents & (EPOLLERR|EPOLLHUP|EPOLLRDHUP)) {
				//客户端关闭，服务器有一个写事件时成立
				--connect->throw_send_count;
			}
			else {
				(this->*(connect->write_handler))(connect);
			}
		}
	}
	//LogStderr(0, "事件处理完毕");
	return 1;
}

//通知专门用于处理的线程发送待发送消息队列中的消息
void *MySocket::ServerSendQueueThread(void* thread_data) {
	ThreadItem* thread = static_cast<ThreadItem*>(thread_data);
	MySocket *socket_obj = thread->this_;
	int err;
	std::list<char *>::iterator pos,pos2,pos_end;

	char *msg_buf;
	LPSTRUC_MSG_HEADER msg_header;
	LPCOMM_PKG_HEADER  pkg_header;
	lpconnection_t     connect;
	unsigned short     tmp;
	ssize_t            send_size;
	
	//LogStderr(0, "进入到开始发生消息的函数中");
	MyMemory *memory = MyMemory::GetInstance();
	while(stop_event == 0) {
		if(sem_wait(&socket_obj->sem_event_send_queue_) == -1) {
			if(errno != EINTR) {
				LogStderr(errno, "MySocket::ServerSendQueueThread()中sem_wait(&socket_obj->sem_event_send_queue_)失败.");
			}
		}
		//LogStderr(0, "收到发送消息进入队列的通知,开始做发送前的准备.");
		if(stop_event != 0) {
			break;
		}
		if(socket_obj->send_msg_queue_count_ > 0) {
			//LogStderr(0,"信号量来的时候，是否消息队列中显示正确");
			err = pthread_mutex_lock(&socket_obj->send_message_queue_mutex_);
			if(err != 0) {
				LogStderr(err, "MySocket::ServerSendQueueThread()中pthread_mutex_lock()失败,返回错误码为%d!", err);
			}
			pos = socket_obj->msg_send_queue_.begin();
			pos_end = socket_obj->msg_send_queue_.end();
			
			int size = socket_obj->send_msg_queue_count_;
			//LogStderr(0, "当前消息队列中的个数%d",size);
			while(pos != pos_end) {
				msg_buf = (*pos);	
				msg_header = (LPSTRUC_MSG_HEADER)msg_buf;
				pkg_header = (LPCOMM_PKG_HEADER)(msg_buf + socket_obj->msg_head_len_);
				connect = msg_header->connect;
				if(connect->current_sequence != msg_header->current_sequence) {

					pos2 = pos;
					pos++;
					socket_obj->msg_send_queue_.erase(pos2);
					--socket_obj->send_msg_queue_count_;

					memory->FreeMemory(msg_buf);
					continue;
				}

				if(connect->throw_send_count > 0) {
					pos++;
					continue;
				}

				--connect->send_count;

				connect->send_mem_pointer = msg_buf;
				pos2 = pos;
				pos++;
				socket_obj->msg_send_queue_.erase(pos2);
				--socket_obj->send_msg_queue_count_;
				connect->send_buf = (char *)pkg_header;
				tmp = ntohs(pkg_header->pkg_len);
				connect->send_len = tmp;
			

				send_size = socket_obj->SendProc(connect, connect->send_buf, connect->send_len);
				if(send_size > 0) {
					if(send_size == connect->send_len) {
						memory->FreeMemory(connect->send_mem_pointer);
						connect->send_mem_pointer = NULL;
						connect->throw_send_count = 0;
					}
					else {
						connect->send_buf = connect->send_buf + send_size;
						connect->send_len = connect->send_len - send_size;

						++connect->throw_send_count;

						if(socket_obj->EpollOperEvent(connect->fd, EPOLL_CTL_MOD, EPOLLOUT, 0, connect) == -1) {
								LogStderr(errno, "MySocket::ServerSendQueueThread()中EpollOperEvent()失败.");
						}
					}
					continue;
				}
				else if(send_size == 0) {
					memory->FreeMemory(connect->send_mem_pointer);
					connect->send_mem_pointer = NULL;
					connect->throw_send_count = 0;
					continue;
				}
				else if(send_size == -1) {
					++connect->throw_send_count;
					if(socket_obj->EpollOperEvent(connect->fd, EPOLL_CTL_MOD, EPOLLOUT, 0, connect) == -1) {
						LogStderr(errno, "MySocket::ServerSendQueueThread()中EpollOperEvent()失败.");
					}
					continue;
				}
				else {
					memory->FreeMemory(connect->send_mem_pointer);
					connect->send_mem_pointer = NULL;
					connect->throw_send_count = 0;
					continue;
				}
			}//end while
			err = pthread_mutex_unlock(&socket_obj->send_message_queue_mutex_);
			if(err != 0) {
				LogStderr(err, "MySocket::ServerSendQueueThread()中pthread_mutex_unlock()失败,返回的错误码为%d!", err);
			}
		}//end if
	}//end while
	return (void*)0;
}
